package com.core.image.core.impl.async;

import com.core.image.core.IoArgs;
import com.core.image.core.SendDispatcher;
import com.core.image.core.SendPacket;
import com.core.image.core.Sender;
import com.core.image.utils.CloseUtils;

import java.io.IOException;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * describe:
 *
 * @author 库琦
 * @date 2020/01/24
 */
public class AsyncSendDispatcher implements
        SendDispatcher,IoArgs.IoArgsEventProcessor,AsyncPacketReader.PacketProvider{
    private final Sender sender;
    private final Queue<SendPacket> queue = new ConcurrentLinkedQueue<>();
    private final AtomicBoolean isSending = new AtomicBoolean();
    private final AtomicBoolean isClosed = new AtomicBoolean(false);
    private AsyncPacketReader reader =new  AsyncPacketReader(this);
    public AsyncSendDispatcher(Sender sender){
            this.sender =sender;
            this.sender.setSendListener(this);
    }
    @Override
    public void send(SendPacket packet) {
            queue.offer(packet);
             requestSend();
    }
    @Override
    public SendPacket takePacket(){
        SendPacket packet = queue.poll();
            if (packet == null){
                return null;
        }
        if (packet.isCanceled()){
            return takePacket();
        }
             return packet;
    }

    @Override
    public void completedPacket(SendPacket packet, boolean isSuccessed) {
        CloseUtils.close(packet);
    }


    /**
     * 请求进行网络发送
     */
    private void requestSend() {
        synchronized (isSending){
            if (isSending.get()|| isClosed.get()){
                return;
            }
            if (reader.requestTakePacket()){
                try {
                    boolean isSuccess = sender.postSendAsync();
                    if(isSuccess){
                        isSending.set(true);
                    }
                }catch (Exception e){
                    e.printStackTrace();
                    closeAndNotify();
                }
            }
        }

    }
    private void closeAndNotify(){
        CloseUtils.close(this);
    }

    @Override
    public void cancel(SendPacket packet) {
        boolean ret;
        ret= queue.remove(packet);
        if (ret){
            packet.cancel();
            return;
        }
        reader.cancel(packet);

    }

    @Override
    public void sendHeartbeat() {
        if (queue.size() > 0){
            return;
        }
        if (reader.requestSendHeartbeatFrame()){
            requestSend();
        }
    }


    @Override
    public void close() throws IOException {
        if(isClosed.compareAndSet(false,true)){
            //异常关闭操作。
            reader.close();
            queue.clear();
            synchronized (isSending){
                isSending.set(false);
            }
        }
    }

    @Override
    public IoArgs provideIoArgs() {
            return isClosed.get()? null : reader.fillData();
    }

    @Override
    public void onConsumeFailed(IoArgs ioArgs, Exception e) {
           e.printStackTrace();
           synchronized (isSending){
               isSending.set(false);
           }
           requestSend();
    }
    @Override
    public void onConsumeComplete(IoArgs ioArgs) {
        synchronized (isSending){
            isSending.set(false);
        }
        //继续发送当前包
        if (reader.requestTakePacket()){
            requestSend();
        }
    }
}
